Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@rationai/mlkit/data/shard_parquet.py`:
- Line 7: The code currently uses os.makedirs(output_dir, exist_ok=True) and
deterministic shard filenames like shard_00000.parquet which lets reruns
silently overwrite existing shards; change this by adding a pre-check for
existing shard files in output_dir (e.g., glob for "shard_*.parquet") and
fail/raise if any are found or create a uniquely-named output directory
(timestamp/suffix) instead, and remove or set exist_ok=False on
os.makedirs(output_dir, ...) so the guard triggers; apply the same protection
around the shard-writing loop that emits shard_*.parquet (lines that write
shards) to prevent accidental overwrites.
- Line 6: Validate sharding parameters at the start of shard_parquet: check that
rows_per_shard and row_group_size are positive integers (greater than 0) and
raise a clear ValueError if not; include the parameter name and offending value
in the error message so callers immediately see which argument is invalid and
why before any I/O or processing occurs.
- Around line 18-41: The loop over parquet_file.iter_batches currently can skip
writer.close() if iter_batches or writer.write_batch raises; wrap the entire
batching logic in a try/finally (or use a context manager for pq.ParquetWriter)
so that writer.close() is always called when writer is not None; ensure you
reference the same writer variable used when creating
out_path/shard_{shard_idx:05d}.parquet and that current_shard_rows/shard_idx
semantics remain unchanged so partial shards are closed in the finally block on
any exception.
- Around line 27-36: The code currently writes whole batches with
writer.write_batch(batch) then checks current_shard_rows, allowing a shard to
exceed rows_per_shard by a full batch; change write logic in the loop that
handles `batch` so that before calling writer.write_batch you check whether
batch.num_rows + current_shard_rows > rows_per_shard and, if so, split the batch
into smaller pieces and write only up to the remaining rows for the current
shard, close the writer (writer.close(); writer = None), increment shard_idx and
reset current_shard_rows, then create/open a new writer and continue writing the
rest of the batch until the entire batch is consumed; use the existing symbols
`batch`, `batch.num_rows`, `current_shard_rows`, `rows_per_shard`,
`writer.write_batch`, `writer.close`, and `shard_idx` to implement the looped
slicing/writing until all rows are written and no shard exceeds rows_per_shard.